//
//  Synchronized publisher
//

package main

import (
	"fmt"
	"os"
	"zmq"
)

const (
	SUBSCRIBERS_EXPECTED = 10
)

func zmq_main() os.Error {
	context, err := zmq.Init(1)
	if err != nil {
		return err
	}
	defer context.Close()

	// Socket to talk to clients
	publisher, err := context.Socket(zmq.PUB)
	if err != nil {
		return err
	}
	defer publisher.Close()
	err = publisher.Bind("tcp://*:5561")
	if err != nil {
		return err
	}

	// Socket to receive signals
	syncservice, err := context.Socket(zmq.REP)
	if err != nil {
		return err
	}
	defer syncservice.Close()
	err = syncservice.Bind("tcp://*:5562")
	if err != nil {
		return err
	}

	// Get synchronization from subscribers
	for i := 0; i < SUBSCRIBERS_EXPECTED; i++ {
		_, err := syncservice.Recv(0)
		if err != nil {
			return err
		}
		err = syncservice.Send(nil, 0)
		if err != nil {
			return err
		}
	}

	for i := 0; i < 1000000; i++ {
		err = publisher.Send([]byte("Rhubarb"), 0)
		if err != nil {
			return err
		}
	}
	err = publisher.Send([]byte("END"), 0)
	if err != nil {
		return err
	}
	return nil
}

var errch = make(chan os.Error)

func main() {
	fmt.Println("Synchronized publisher")
	exit := make(chan bool)
	go func() {
		for {
			err := <-errch
			if err != nil {
				fmt.Println(err)
				os.Exit(1)
			}
		}
	}()
	go func() {
		err := zmq_main()
		if err != nil {
			errch <- err
		} else {	
			exit <- true
		}
	}()
	<-exit
}